/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.eagle.alert.engine.runner;

import org.apache.eagle.alert.metric.IMetricSystem;
import org.apache.eagle.alert.metric.MetricSystem;
import backtype.storm.metric.api.IMetricsConsumer;
import backtype.storm.task.IErrorReporter;
import backtype.storm.task.TopologyContext;
import com.codahale.metrics.Gauge;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import com.typesafe.config.ConfigParseOptions;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * Share same metric system.
 */
public class StormMetricConsumer implements IMetricsConsumer {
    public static final Logger LOG = LoggerFactory.getLogger(StormMetricConsumer.class);
    private String topologyName;
    private IMetricSystem metricSystem;
    private String topologyId;

    @SuppressWarnings( {"serial", "rawtypes"})
    @Override
    public void prepare(Map stormConf, Object registrationArgument, TopologyContext context, IErrorReporter errorReporter) {
        Config config = ConfigFactory.parseString((String) registrationArgument, ConfigParseOptions.defaults());
        topologyName = config.getString("appId");
        topologyId = context.getStormId();
        metricSystem = MetricSystem.load(config);
        metricSystem.tags(new HashMap<String, Object>() {
            {
                put("appId", topologyName);
                put("stormId", topologyId);
            }
        });
        metricSystem.start();
    }

    @SuppressWarnings( {"unchecked", "rawtypes"})
    @Override
    public void handleDataPoints(TaskInfo taskInfo, Collection<DataPoint> dataPoints) {
        synchronized (metricSystem) {
            List<String> metricList = new LinkedList<>();
            for (DataPoint dataPoint : dataPoints) {
                if (dataPoint.value instanceof Map) {
                    Map<String, Object> values = (Map<String, Object>) dataPoint.value;
                    for (Map.Entry<String, Object> entry : values.entrySet()) {
                        String metricName = buildMetricName(taskInfo, dataPoint.name, entry.getKey());
                        metricList.add(metricName);
                        Gauge gauge = metricSystem.registry().getGauges().get(metricName);
                        if (gauge == null) {
                            LOG.info("Register metric {}", metricName);
                            gauge = new DataPointGauge(entry.getValue());
                            metricSystem.registry().register(metricName, gauge);
                        } else {
                            ((DataPointGauge) gauge).setValue(entry.getValue());
                        }
                    }
                } else {
                    String metricName = buildMetricName(taskInfo, dataPoint.name);
                    metricList.add(metricName);
                    LOG.info("Register metric {}", metricName);
                    Gauge gauge = metricSystem.registry().getGauges().get(metricName);
                    if (gauge == null) {
                        LOG.info("Register metric {}", metricName);
                        gauge = new DataPointGauge(dataPoint.value);
                        metricSystem.registry().register(metricName, gauge);
                    } else {
                        ((DataPointGauge) gauge).setValue(dataPoint.value);
                    }
                }
            }
            metricSystem.registry().removeMatching((name, metric) -> metricList.indexOf(name) < 0);
            metricSystem.report();
            metricSystem.registry().getGauges().values().forEach((gauge -> {
                if (gauge instanceof DataPointGauge) {
                    ((DataPointGauge) gauge).reset();
                }
            }));
            LOG.info("Reported {} metric data points from {} [{}]", dataPoints.size(), taskInfo.srcComponentId, taskInfo.srcTaskId);
        }
    }

    private class DataPointGauge implements Gauge<Object> {
        private Object value;

        public DataPointGauge(Object initialValue) {
            this.value = initialValue;
        }

        @Override
        public Object getValue() {
            return value;
        }

        public void setValue(Object value) {
            this.value = value;
        }

        public void reset() {
            this.value = 0;
        }
    }

    private String buildMetricName(TaskInfo taskInfo, String... name) {
        return String.join(".", StringUtils.join(name, ".").replace("/", "."), taskInfo.srcComponentId, taskInfo.srcTaskId + "");
    }

    @Override
    public void cleanup() {
        metricSystem.stop();
    }
}